package com.junxonline.instant.terminal.netty.handler;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;
import com.junxonline.instant.common.constant.*;
import com.junxonline.instant.common.entity.terminal.websocket.AuthFrame;
import com.junxonline.instant.common.entity.terminal.websocket.MessageFrame;
import com.junxonline.instant.common.entity.terminal.websocket.PingFrame;
import com.junxonline.instant.common.entity.terminal.websocket.PongFrame;
import com.junxonline.instant.common.enumeration.TerminalCommandEnum;
import com.junxonline.instant.common.util.common.RedisUtils;
import com.junxonline.instant.common.util.terminal.NettyChannelUtils;
import com.junxonline.instant.common.util.terminal.TokenUtils;
import com.junxonline.instant.dao.entity.frame.*;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketFrame;
import io.netty.util.concurrent.ScheduledFuture;
import lombok.extern.log4j.Log4j2;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;

/**
 * Websocket帧接收处理器
 *
 * @author JunX
 * @date 2021-09-08
 */
@Log4j2
@Component
@Sharable
public class WebSocketFrameHandler extends SimpleChannelInboundHandler<WebSocketFrame> {

    @Autowired
    private TokenUtils tokenUtils;

    @Autowired
    private RedisUtils redisUtils;

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Override
    protected void channelRead0(ChannelHandlerContext context, WebSocketFrame frame) {
        if (frame instanceof CloseWebSocketFrame) {
            log.info("收到客户端 -> " + context.channel().remoteAddress() + " 关闭帧 -> " + frame);
            context.channel().close();
            log.info("客户端 -> " + context.channel().remoteAddress() + " 主动断开了连接");
        } else if (frame instanceof TextWebSocketFrame) {
            try {
                JSONObject json = JSON.parseObject(((TextWebSocketFrame) frame).text());
                String serialId = json.getString("serialId");
                Integer messageType = json.getInteger("type");
                JSONObject content = json.getJSONObject("content");
                if (messageType.equals(TerminalCommandEnum.HEART_BEAT.getValue())) {
                    // Ping/Pong消息处理
                    log.info("收到客户端 -> " + context.channel().remoteAddress() + " 心跳帧 -> " + ((TextWebSocketFrame) frame).text());
                    PingFrame pingFrame = JSON.toJavaObject(content, PingFrame.class);
                    pingPongMessageHandler(context, serialId, messageType, pingFrame);
                }
                // 认证消息处理
                else if (messageType.equals(TerminalCommandEnum.AUTH.getValue())) {
                    log.info("收到客户端 -> " + context.channel().remoteAddress() + " 认证消息 -> " + ((TextWebSocketFrame) frame).text());
                    AuthFrame authFrame = JSON.toJavaObject(content, AuthFrame.class);
                    authMessageHandler(context, serialId, messageType, authFrame);
                }
                // 拉取音量消息处理
                else if (messageType.equals(TerminalCommandEnum.QUERY_VOLUME.getValue())) {
                    log.info("收到客户端 -> " + context.channel().remoteAddress() + " 拉取音量消息 -> " + ((TextWebSocketFrame) frame).text());
                    QueryVolumeFrame queryVolumeFrame = JSON.toJavaObject(content, QueryVolumeFrame.class);
                    queryVolumeMessageHandler(context, serialId, messageType, queryVolumeFrame);
                }
                // 拉取显示方式消息处理
                else if (messageType.equals(TerminalCommandEnum.QUERY_DISPLAY_TYPE.getValue())) {
                    log.info("收到客户端 -> " + context.channel().remoteAddress() + " 拉取显示方式消息 -> " + ((TextWebSocketFrame) frame).text());
                    QueryDisplayTypeFrame queryDisplayTypeFrame = JSON.toJavaObject(content, QueryDisplayTypeFrame.class);
                    queryDisplayTypeMessageHandler(context, serialId, messageType, queryDisplayTypeFrame);
                }
                // 拉取截屏消息处理
                else if (messageType.equals(TerminalCommandEnum.PULL_SCREENSHOT.getValue())) {
                    log.info("收到客户端 -> " + context.channel().remoteAddress() + " 拉取截屏消息 -> " + ((TextWebSocketFrame) frame).text());
                    PullScreenshotFrame pullScreenshotFrame = JSON.toJavaObject(content, PullScreenshotFrame.class);
                    pullScreenshotMessageHandler(context, serialId, messageType, pullScreenshotFrame);
                } else {
                    log.info("收到客户端 -> " + context.channel().remoteAddress() + " 消息 -> " + ((TextWebSocketFrame) frame).text());
                }
            } catch (JSONException e) {
                log.error("收到非JSON格式的数据");
            } catch (Exception e) {
                log.error("读取消息时发生了错误");
                e.printStackTrace();
            }
        } else {
            throw new UnsupportedOperationException(frame.getClass().getName() + " 不支持的帧类型");
        }

    }

    /**
     * Ping/Pong消息处理
     *
     * @param context     上下文
     * @param serialId    消息ID
     * @param messageType 消息类型
     */
    private void pingPongMessageHandler(ChannelHandlerContext context, String serialId, Integer messageType, PingFrame frame) {
        String terminalCode = NettyChannelUtils.getTerminalCode(context.channel().id());
        // 将终端编号与状态发送至Redis消息队列中处理
        Map<String, Object> dataMap = new HashMap<>();
        dataMap.put("terminalCode", terminalCode);
        dataMap.put("status", frame.getStatus());
        redisUtils.convertAndSend(RedisMqChannelConstant.KEY_REDIS_MSG_CHANNEL_TERMINAL_STATUS, JSON.toJSONString(dataMap));
        MessageFrame<PongFrame> messageFrame = new MessageFrame<>(serialId, DateUtil.current(), messageType, new PongFrame(CommonConstant.SUCCESS));
        String jsonMessage = JSONUtil.toJsonStr(messageFrame);
        context.channel().writeAndFlush(new TextWebSocketFrame(jsonMessage));
        log.info("已回复客户端 -> " + context.channel().remoteAddress() + " 心跳帧 -> " + jsonMessage);
    }

    /**
     * 认证消息处理
     *
     * @param context     上下文
     * @param serialId    消息ID
     * @param messageType 消息类型
     * @param frame       消息内容
     */
    private void authMessageHandler(ChannelHandlerContext context, String serialId, Integer messageType, AuthFrame frame) {
        // 解析消息
        String terminalCode = frame.getTerminalCode();
        String sign = frame.getSign();
        String token = frame.getToken();

        // TODO 校验签名

        Integer authStatus = TerminalAuthStatusConstant.FAIL;
        // 校验token
        String redisToken = (String) redisUtils.hget(RedisConstant.KEY_TERMINAL_AUTH + terminalCode, TerminalInfoConstant.KEY_TOKEN);
        if (StrUtil.isNotBlank(redisToken) && token.equals(redisToken) && tokenUtils.verify(token, terminalCode)) {
            Channel channel = NettyChannelUtils.getChannel(terminalCode);
            if (channel == null) {
                authStatus = TerminalAuthStatusConstant.SUCCESS;
            } else {
                authStatus = TerminalAuthStatusConstant.EXIST;
            }
        }

        // 取消主动关闭连接定时任务
        cancelDisconnectionSchedule(context);
        log.info("取消了客户端 -> {} 终端编号 -> {} 主动关闭连接的定时任务", context.channel().remoteAddress(), terminalCode);
        // 清除Redis中的认证token
        redisUtils.del(RedisConstant.KEY_TERMINAL_AUTH + terminalCode);
        log.info("删除了客户端 -> {} Redis中的认证Token", context.channel().remoteAddress());
        // 认证通过
        if (authStatus == TerminalAuthStatusConstant.SUCCESS) {
            // 将终端编号与channel绑定加入Channel容器中
            NettyChannelUtils.addChannel(terminalCode, context.channel());
            responseAuthResult(context, serialId, messageType, TerminalAuthStatusConstant.SUCCESS);
            log.info("客户端 -> {} 认证通过", context.channel().remoteAddress());
            // 认证通过后需要执行的业务
            authSuccessCallback(context, terminalCode, serialId, messageType);
        } else {
            responseAuthResult(context, serialId, messageType, authStatus);
            // 认证失败回复消息后关闭连接
            context.channel().close();
            log.info("客户端 -> {} 认证失败", context.channel().remoteAddress());
        }
    }

    /**
     * 拉取音量消息处理
     *
     * @param context     上下文
     * @param serialId    消息ID
     * @param messageType 消息类型
     * @param frame       消息内容
     */
    private void queryVolumeMessageHandler(ChannelHandlerContext context, String serialId, Integer messageType, QueryVolumeFrame frame) {
        String terminalCode = NettyChannelUtils.getTerminalCode(context.channel().id());
        Map<String, Object> dataMap = new HashMap<>();
        dataMap.put("terminalCode", terminalCode);
        dataMap.put("volume", frame.getVolume());
        // 将消息发送至消息队列中处理
        String destination = TerminalCommandMqConstant.TOPIC_QUERY_COMMAND_RESPONSE + ":" + TerminalCommandMqConstant.TAG_QUERY_VOLUME_RESPONSE;
        rocketMQTemplate.syncSend(destination, MessageBuilder.withPayload(JSON.toJSONString(dataMap)).build());
    }

    /**
     * 拉取显示方式消息处理
     *
     * @param context     上下文
     * @param serialId    消息ID
     * @param messageType 消息类型
     * @param frame       消息内容
     */
    private void queryDisplayTypeMessageHandler(ChannelHandlerContext context, String serialId, Integer messageType, QueryDisplayTypeFrame frame) {
        String terminalCode = NettyChannelUtils.getTerminalCode(context.channel().id());
        Map<String, Object> dataMap = new HashMap<>();
        dataMap.put("terminalCode", terminalCode);
        dataMap.put("displayType", frame.getDisplayType());
        // 将消息发送至消息队列中处理
        String destination = TerminalCommandMqConstant.TOPIC_QUERY_COMMAND_RESPONSE + ":" + TerminalCommandMqConstant.TAG_QUERY_DISPLAY_TYPE_RESPONSE;
        rocketMQTemplate.syncSend(destination, MessageBuilder.withPayload(JSON.toJSONString(dataMap)).build());
    }

    /**
     * 拉取截屏消息处理
     *
     * @param context     上下文
     * @param serialId    消息ID
     * @param messageType 消息类型
     * @param frame       消息内容
     */
    private void pullScreenshotMessageHandler(ChannelHandlerContext context, String serialId, Integer messageType, PullScreenshotFrame frame) {
        String terminalCode = NettyChannelUtils.getTerminalCode(context.channel().id());
        Map<String, Object> dataMap = new HashMap<>();
        dataMap.put("terminalCode", terminalCode);
        dataMap.put("url", frame.getUrl());
        // 将消息发送至消息队列中处理
        String destination = TerminalCommandMqConstant.TOPIC_QUERY_COMMAND_RESPONSE + ":" + TerminalCommandMqConstant.TAG_PULL_SCREENSHOT_RESPONSE;
        rocketMQTemplate.syncSend(destination, MessageBuilder.withPayload(JSON.toJSONString(dataMap)).build());
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext context) throws Exception {
        context.channel().flush();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext context, Throwable cause) {
        String terminalCode = NettyChannelUtils.getTerminalCode(context.channel().id());
        Map<String, Object> dataMap = new HashMap<>();
        dataMap.put("terminalCode", terminalCode);
        dataMap.put("status", CommonConstant.FAIL);
        // 将消息发送至Redis消息队列中处理
        redisUtils.convertAndSend(RedisMqChannelConstant.KEY_REDIS_MSG_CHANNEL_TERMINAL_STATUS, JSON.toJSONString(dataMap));
        // 移除容器中的Channel
        NettyChannelUtils.removeChannel(context.channel());
        context.channel().close();
    }

    /**
     * 取消主动关闭连接的定时任务
     *
     * @param context 上下文
     */
    private void cancelDisconnectionSchedule(ChannelHandlerContext context) {
        // 取消主动关闭连接的定时任务
        ScheduledFuture<ChannelFuture> closeFuture = context.channel().attr(HandshakeEventHandler.DISCONNECTION_SCHEDULE_FUTURE).getAndSet(null);
        if (closeFuture != null) {
            closeFuture.cancel(true);
        }
    }

    /**
     * 回复认证结果
     *
     * @param context     上下文
     * @param serialId    消息ID
     * @param messageType 消息类型
     * @param authStatus  认证结果
     */
    private void responseAuthResult(ChannelHandlerContext context, String serialId, Integer messageType, Integer authStatus) {
        MessageFrame<PongFrame> messageFrame = new MessageFrame<>(serialId, DateUtil.current(), messageType, new PongFrame(authStatus));
        String jsonMessage = JSONUtil.toJsonStr(messageFrame);
        context.channel().writeAndFlush(new TextWebSocketFrame(jsonMessage));
        log.info("已回复客户端 -> " + context.channel().remoteAddress() + " 认证结果 -> " + jsonMessage);
    }

    /**
     * 认证
     *
     * @param context      上下文
     * @param terminalCode 终端编号
     * @param serialId     消息ID
     * @param messageType  消息类型
     */
    private void authSuccessCallback(ChannelHandlerContext context, String terminalCode, String serialId, Integer messageType) {
        // 连接日志插入 将终端编号与状态发送至Redis消息队列中处理
        Map<String, Object> terminalStatusMap = new HashMap<>();
        terminalStatusMap.put("terminalCode", terminalCode);
        terminalStatusMap.put("type", TerminalConnectTypeConstant.ONLINE);
        redisUtils.convertAndSend(RedisMqChannelConstant.KEY_REDIS_MSG_CHANNEL_LOG_DEVICE_CONNECT, JSON.toJSONString(terminalStatusMap));
    }

}
